package club.monkeywood.ad.dmp.tool

import ch.hsr.geohash.GeoHash
import club.monkeywood.ad.dmp.util.{BaiduGeoApi, JedisPools}
import org.apache.commons.lang.StringUtils
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

/**
	* 创建商圈库
  * 用来抽取日志字段中的经纬度，并请求百度的api 获取到商圈信息
	* 将日志中的坐标转换为geohash
	* 将(key=geohash,value=商圈信息)存入redis
  */
object ExtractLatLong2Business {

    def main(args: Array[String]): Unit = {

        // 0 校验参数个数
        if (args.length != 1) {
            println(
                """
                  |cn.dmp.tools.ExtractLatLong2Business
                  |参数：
                  | inputPath
                """.stripMargin)
            sys.exit()
        }

        // 1 接受程序参数
        val Array(inputPath) = args

      // 2 创建sparkconf->SparkSession
      val sparkConf = new SparkConf()
      sparkConf.setAppName(s"${this.getClass.getSimpleName}")
      sparkConf.setMaster("local[*]")
      //使用KryoSerializer更快
      // RDD 序列化到磁盘 worker与worker之间的数据传输
      sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

      val ss = SparkSession.builder()
        .appName(s"${this.getClass.getSimpleName}")
        .config(sparkConf)
        .getOrCreate()

      ss.read.parquet(inputPath)
//          .where(TagsUtils.testCdt).show()
        .select("lat", "log")
        //中国经度范围:73-136
        //中国纬度范围：3-54
        .where("lat > 3 and lat < 54 and log > 73 and log < 136").distinct()
        .foreachPartition(itr => {
            // jedis客户端
            val jedis = JedisPools.getConnection()

            itr.foreach(row => {
                val lat = row.getAs[String]("lat")
                val longs = row.getAs[String]("log")

                //获取geohash值
                //geohash编码长度：8位
                val geoHashCode = GeoHash.withCharacterPrecision(lat.toDouble, longs.toDouble, 8).toBase32
                val business = BaiduGeoApi.getBusiness(longs,lat)
                if (StringUtils.isNotEmpty(business)) {
                    jedis.set(geoHashCode, business)
                }

            })
            jedis.close()
        })

      ss.stop()
    }

}
